package com.weishi.project.speedometer.domain;

import com.weishi.framework.config.BeanContext;
import com.weishi.project.speedometer.service.ISpeedometerService;
import com.weishi.project.speedometer.vo.DevInfo;
import org.eclipse.paho.client.mqttv3.*;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

public class EmqMqttClient {

    private ScheduledExecutorService pool = Executors.newScheduledThreadPool(5);

    private String clientId = MqttClient.generateClientId();
    private String serviceUrl = "tcp://je9de665.cn-shenzhen.emqx.cloud:15473";
    private String username = "xskj";
    private String password = "xs@123";

    /**
     * 通知队列
     */
    private Map<String,Integer> noticeQueue = new HashMap<>();

    private ISpeedometerService speedometerService;

    private static EmqMqttClient emqMqttClient;

    private EmqMqttClient(){
        this.speedometerService = BeanContext.getApplicationContext().getBean(ISpeedometerService.class);
    }

    public static EmqMqttClient getInstance(){
        if(emqMqttClient==null){
            emqMqttClient = new EmqMqttClient();
        }
        return emqMqttClient;
    }

    /**
     * 添加设备到通知队列
     * @param devId
     * @param channelId
     */
    public void setNoticeQueue(String devId,int channelId) {
        this.noticeQueue.put(devId,channelId);
    }

    /**
     * 初始化客户端
     */
    public boolean init(String service,String port,String user,String pass,String clientId,String topic) {
        this.clientId = clientId;
        this.serviceUrl = service+":"+port;
        this.username = user;
        this.password = pass;
        try {
            //创建监听客户端
            createMqttClient(topic,new MqttGpsCallBack());
            createMqttClient(topic,new MqttAccelerationCallBack());
            createMqttClient(topic,new MqttHeartCallBack());
            createMqttClient(topic,new WitCallBack());
//            createMqttClient("gps_topic",new SimpleMqttCallBack());

            //客户端发送数据
            sendMessage(clientId,topic);
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }

    private void createMqttClient(String topic, MqttCallBackCommon mqttCallback){
        try {
            MqttClient client = new MqttClient(this.serviceUrl, topic+"_"+clientId+"_"+mqttCallback.getTopic());
            MqttConnectOptions options = new MqttConnectOptions();
            // 设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，设置为true表示每次连接到服务器都以新的身份连接
            options.setCleanSession(false);
            // 设置连接的用户名
            options.setUserName(this.username);
            // 设置连接的密码
            options.setPassword(this.password.toCharArray());
            // 设置超时时间 单位为秒
            options.setConnectionTimeout(3);
            // 设置会话心跳时间 单位为秒 服务器会每隔1.5*10秒的时间向客户端发送个消息判断客户端是否在线，但这个方法并没有重连的机制
            options.setKeepAliveInterval(10);
            //设置断开后重新连接
            options.setAutomaticReconnect(true);
            client.setCallback(mqttCallback);
            client.connect(options);
            client.subscribe(topic+":"+mqttCallback.getTopic());
//            client.subscribe(mqttCallback.getTopic());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void sendMessage(String clientId,String topic) throws Exception{
        MqttConnectOptions options = new MqttConnectOptions();
        options.setCleanSession(false);
        options.setUserName(this.username);
        options.setPassword(this.password.toCharArray());
        options.setConnectionTimeout(10);
        options.setKeepAliveInterval(20);
        options.setAutomaticReconnect(true);
        MqttClient client = new MqttClient(this.serviceUrl, topic+"_"+clientId + "_" + "notice");
        client.connect(options);
        pool.execute(new Runnable() {
            MqttMessage message = new MqttMessage();
            @Override
            public void run() {
                while (true){
                    try {
                        for(String devId : noticeQueue.keySet()){
                            if(noticeQueue.get(devId)>0){
                                DevInfo devInfo = speedometerService.selectRealTimeDataByDevId(devId);
                                devInfo.setRealTime();
                                message.setPayload(devInfo.toString().getBytes());
                                try {
                                    client.publish("devInfo_"+devId, message);
                                } catch (MqttException e) {

                                }
                            }
                        }
                        Thread.sleep(1000);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                }
            }
        });
    }
}
